package org.itew.mymapreduce.core;

import java.util.Collection;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.BlockingQueue;

import org.itew.mymapreduce.io.Writable;

/**
 * 该类是一个reduce任务的抽象，它负责将结果收集
 * @author JiLu
 * 
 */
public class BasicReducerRunner<IK,IV,OK,OV> implements Runnable{
	
	private final Collection<DataCollection<IK,IV>> mappedDataCollections;
	private final Reducer<IK,IV,OK,OV> reducer;
	private final Writable<OK,OV> writer;
	private final BlockingQueue<Set<IK>> keySetQueue;
	private final Set<IK> endOfKeySet;
	
	public BasicReducerRunner(Collection<DataCollection<IK,IV>> mappedDataCollections,Reducer<IK,IV,OK,OV> reducer,Writable<OK,OV> writer,BlockingQueue<Set<IK>> keySetQueue,Set<IK> endOfKeySet){
		this.mappedDataCollections = mappedDataCollections;
		this.reducer = reducer;
		this.writer = writer;
		this.keySetQueue = keySetQueue;
		this.endOfKeySet = endOfKeySet;
	}

	@Override
	public void run() {
		try {
			
			while(true){
				
				Set<IK> keySet = keySetQueue.take();
				
				if(keySet==endOfKeySet){
					keySetQueue.put(endOfKeySet);
					break;
				}
				
				for(IK key : keySet){
					Iterator<DataCollection<IK,IV>> mergedDataCollectionsIterator = mappedDataCollections.iterator();
					
					Collection<IV> values = null;
					try{
					while(mergedDataCollectionsIterator.hasNext()){
						if(values ==null)
							values = mergedDataCollectionsIterator.next().getValues(key);
						else{
							
							DataCollection<IK, IV> dc = mergedDataCollectionsIterator.next();
							
							Collection<IV> c = dc.getValues(key);
							
							if(c!=null)
								values.addAll(c);
							
						}
					
					}
					}catch(NullPointerException e){
						System.out.println(values+"@"+key);
					}
					reducer.reduce(key, values, writer);
				}
				
			}
			
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		}finally{
			writer.close();
		}
		
	}
	
}
